package com.gitee.sparta.core.cluster;

import com.gitee.sparta.core.bean.cluster.ClusterCmd;
import com.gitee.sparta.common.IdentityEnum;
import com.gitee.sparta.common.helpers.IpHelper;
import com.gitee.sparta.common.helpers.IsolationHelper;
import com.gitee.sparta.core.bean.cluster.ClusterProtocol;
import com.gitee.sparta.core.bean.cluster.ServerBean;
import com.gitee.sparta.core.bean.event.LeaderEnableEvent;
import com.gitee.sparta.core.cluster.balance.LoadBalance;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.springframework.context.ApplicationEventPublisher;

import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * <p>
 *      集群公共的代码
 * </p>
 *
 * @author zyred
 * @since v 0.0.2
 */
@Slf4j
@AllArgsConstructor
public class ClusterCommonCode {

    private static final int MAX_COUNT = 10;
    private static int tryCount = 0;

    private final RedissonClient client;
    private final LoadBalance loadBalance;
    private final SpartaCoreProperties properties;
    private final ApplicationEventPublisher eventPublisher;


    /**
     * 晋升 leader 方法
     * 方法执行时机: 当集群中 leader 节点宕机或者是网络通讯故障, 所有的 follower 的连接会失败
     * 此时就会触发 follower 晋升 leader 的程序
     *
     * 核心原理: 抢占 redis 锁, 谁抢占成功,谁就有机会成为 leader
     *
     * @param leader        抽象 leader
     * @param follower      抽象 follower
     */
    public void promotionLeader(AbstractClusterLeader leader, AbstractClusterFollower follower) {
        ++ tryCount;
        RLock lock = this.client.getLock(this.properties.getLock());
        // 等待 1000 ms, 没抢占到锁的服务,会一直轮询查找 leader
        long time = 1000;
        try {
            // 抢占成功，将自己变成 leader， 并把自己注册到 serverTab 中
            // 抢占失败，从列表中读取 leader 信息，进行连接
            if (lock.tryLock(time, TimeUnit.MICROSECONDS)) {
                log.info("[Sparta] Get lock, start promotion leader.");
                RMap<Integer, ServerBean> serverTab = this.client.getMap(this.properties.getServerTable());
                ServerBean bean = serverTab.values().stream()
                        .filter(ServerBean::isLeader).findFirst().orElse(null);

                if (Objects.nonNull(bean)) {
                    serverTab.remove(bean.getServerId());
                }

                log.info("[Sparta] Start build leader.");
                leader.buildLeader();
                leader.processorCmd();
                log.info("[Sparta] Build leader success.");

                String addr = IpHelper.getAddress();
                Integer myId = ClusterCache.getServerId();
                serverTab.remove(myId);

                int ctp = this.properties.getCluster().getClusterTransportPort();
                ServerBean selfBean = ClusterCache.getServerInfo(myId)
                        .setLeader(true).setServerPort(ctp)
                        .setServerId(myId).setServerIp(addr);
                serverTab.put(myId, selfBean);

                ClusterCache.putServerInfo(myId, selfBean);
                ClusterCache.setServerIdentity(IdentityEnum.LEADER);
                String isolation = IsolationHelper.isolation(myId);
                ClusterCache.leaderIsolation = isolation;

                this.loadBalance.addNode(isolation, myId);
                this.eventPublisher.publishEvent(new LeaderEnableEvent(selfBean));

                TimeUnit.MILLISECONDS.sleep(time);
                lock.unlock();

                // 通过消息停止线程
                follower.closeSendThread();
                if (!ClusterCache.isNetty()) {
                    Thread.currentThread().interrupt();
                }

                log.info("[Sparta] Promotion leader success, addr: {}, myId: {}, isolation: {}", addr, myId, isolation);
            } else {
                log.info("[Sparta] Not get lock, try connect new leader.");
                // 关闭对旧的 leader 的连接, 进行重新连接新的 leader
                follower.closeForOldLeader();

                RMap<Integer, ServerBean> serverTab = this.client.getMap(this.properties.getServerTable());
                ServerBean bean = serverTab.values().stream()
                        .filter(ServerBean::isLeader).findFirst().orElse(null);

                while (Objects.isNull(bean)) {
                    TimeUnit.MILLISECONDS.sleep(time);
                    serverTab = this.client.getMap(this.properties.getServerTable());
                    bean = serverTab.values().stream()
                            .filter(ServerBean::isLeader).findFirst().orElse(null);
                }

                log.info("[Sparta] Try connect new leader, host : {}, port : {}",
                        bean.getServerIp(), bean.getServerPort());
                //  更改父类的 leader 信息
                follower.buildFollower(bean.getServerIp(), bean.getServerPort());
                ClusterProtocol pro = ClusterProtocol
                        .defaultValue(ClusterCmd.STOP_LEADER_SENDER);
                AbstractClusterLeader.sendQueue(pro);
                log.info("[Sparta] Connect new leader success.");
            }
        } catch (Exception ex) {
            log.error("[Sparta] Promotion leader error, exception : {}", ex.getMessage(), ex);
            if (tryCount < MAX_COUNT) {
                log.info("[Sparta] Try again promotion leader, try count : {}", tryCount);
                this.promotionLeader(leader, follower);
            }
        }
    }


    /**
     * 当前集群是否发生过二次选举
     * 1. 如果服务刚启动,那么 sid 一定会为空, 此时还未向 leader 发送 accept 指令
     * 2. 如果服务启动完毕并且运行了一段时间,发生了集群选举,那么为  follower 的系
     * 统则需要重新注册到 new leader 上, 此时的 sid 一定不为 空, 此时就要告诉
     * new leader, 该 follower 已经注册过了, 不需要 new leader 重新分配 sid
     *
     * @param sid       sid
     */
    public void firstConnectionLeader (Integer sid) {
        ClusterProtocol protocol = ClusterProtocol.defaultValue(ClusterCmd.CHANNEL_ACCEPT);

        if (Objects.nonNull(sid)) {
            RMap<Integer, ServerBean> serverTab = this.client.getMap(this.properties.getServerTable());
            ServerBean bean = serverTab.get(sid);
            if (Objects.nonNull(bean)) {
                protocol.setServerId(bean.getServerId());
                protocol.setCmd(ClusterCmd.NEXT_FOLLOWER);
                protocol.setLeaderSid(ClusterCache.getLeaderSid());
            }
        }
        AbstractClusterFollower.sendQueue(protocol);
    }
}
